package com.swhy

import java.net.URI
import java.util.{Date, UUID}

import com.utils._
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.{SparkConf, SparkContext}
import org.codehaus.jackson.map.ObjectMapper
import org.slf4j.LoggerFactory


object ChgeCustIndxDetail {
  def main(args: Array[String]): Unit = {
    val logger = LoggerFactory.getLogger(this.getClass)
    if (args.length != 3) {
      println(
        """
          |com.swhy.CompanyInfo
          |params:
          | InputDate    刷数日期
          | OutputPath   输出路径
          | SplitNum     文件生成数量
        """.stripMargin)
      sys.exit()
    }
    val conf = new Configuration()
    val Array(inputDt, outputPath, splitNum) = args
    val sparkConf = new SparkConf().setAppName(this.getClass.getName)
    val sc = new SparkContext(sparkConf)
    val hiveContext = new HiveContext(sc)
//改下方的sql为需要装换成jason的sql
    val allcompany =s"""
        |select
        |distinct
        |A.ecif_cd,
        |A.capi_acc,
        |A.cust_no,
        |A.cust_name,
        |nvl(A.mgr_serv,'') as mgr_cd,
        |nvl(A.mgr_dev,'') as mgr_dev,
        |nvl(A.mgr_sign,'') as mgr_sign,
        |nvl(A.brch_org_level_cd_serv,'') as brch_org_level_cd,
        |nvl(A.brch_org_level_cd_dev,'') as brch_org_level_cd_dev,
        |nvl(A.brch_org_level_cd_sign,'') as brch_org_level_cd_sign,
        |nvl(A.brch_org_cd_serv,'') as brch_org_cd,
        |nvl(A.brch_org_cd_dev,'') as brch_org_cd_dev,
        |nvl(A.brch_org_cd_sign,'') as brch_org_cd_sign,
        |A.acc_clas,
        |round(nvl(A.total_asset,0)) as total_asset,
        |nvl(A.net_asset,0)  as net_asset,
        |A.latest_risk_grad,
        |A.cust_grad,
        |A.cust_type,
        |nvl(A.scr_trdy_20,0) as scr_trdy_20 ,
        |nvl(A.net_asset_avg_20,0) as net_asset_avg_20 ,
        |nvl(A.stk_mkt_val,0) as stk_mkt_val,
        |nvl(A.total_mkt_val,0) as total_mkt_val ,
        |nvl(A.prft_loss_y,0) as prft_loss_y ,
        |nvl(A.prft_loss_ly,0) as prft_loss_ly ,
        |nvl(A.prft_loss_ly2,0) as prft_loss_ly2 ,
        |nvl(A.prft_loss_ly3,0) as prft_loss_ly3 ,
        |nvl(A.prft_rate_y,0) as prft_rate_y,
        |nvl(A.prft_rate_ly,0) as prft_rate_ly ,
        |nvl(A.prft_rate_ly2,0) as prft_rate_ly2,
        |nvl(A.prft_rate_ly3,0) as prft_rate_ly3,
        |nvl(A.prft_loss_m,0) as prft_loss_m,
        |nvl(A.fm_mkt_val,0) as fm_mkt_val ,
        |nvl(A.fnd_mkt_val,0) as fnd_mkt_val,
        |nvl(A.bond_mkt_val,0) as bond_mkt_val ,
        |nvl(A.cap_bal,0)  as cap_bal ,
        |nvl(A.fin_prod_val,0) as fin_prod_val,
        |nvl(A.cap_in,0) as cap_in ,
        |nvl(A.cap_out,0) as cap_out,
        |A.fnd_share_bal ,
        |A.l1y_purc_same_fnd,
        |A.l1y_purc_fnd,
        |A.l1y_new_stk_allo,
        |A.bjs_flag,
        |A.marg_flag,
        |A.sgt_flag,
        |A.hgt_flag,
        |A.opt_flag,
        |A.secu_accet_flag,
        |A.pro_investor_flag,
        |A.effect_flag,
        |A.logn_tims_w,
        |A.logn_tims_l4w,
        |A.logn_tims_l1m,
        |A.tm,
        |B.SJ01,
        |B.SJ02,
        |B.SJ03,
        |B.SJ04,
        |B.SJ05,
        |B.SJ06,
        |B.SJ07,
        |B.FX01,
        |B.FX02,
        |B.FX03,
        |B.FX04,
        |if(regexp_replace(F.risk_rating_effct_dt,'-','')<'${inputDt}','1','0') as  FX05,
        |if(F.orig_cert_end_dt<'${inputDt}','1','0') as  FX06,
        |B.SJ08,
        |B.SJ09,
        |B.SJ10,
        |B.SJ11,
        |round(nvl(C.model_score,0),4) as model_score,
        |C.loss_reason,
        |nvl(E.ylj_cust_logo,'0') as ylj_cust_logo,
        |E.person_pension_cust_flag as pension_appoint_open_sign,
        |if(trim(regexp_replace(E.person_pension_cust_age_groups,',',''))='3','1','0') as ind_pension_age_segment_3,
        |if(trim(regexp_replace(E.person_pension_cust_age_groups,',',''))='4','1','0') as ind_pension_age_segment_4,
        |if(trim(E.usable invest asset)='A','1','0') as inv_assets_for_risk_assess_a,
        |if(trim(E.usable invest asset)='B','1','0') as inv_assets_for_risk_assess_b,
        |if(trim(E.usable invest asset) in ('C','D'),'1','0') as inv_assets_for_risk_assess_cd,
        |if(trim(E.prod_match_result)='2','1','0') as prod_sub_match_result_12,
        |if(trim(E.prod_match_result)='3','1','0') as prod_sub_match_result_3,
        |E.tg_prod_cust_flag,
        |if（regexp_replace(E.tg_prod_hold_flag,',','')='1','1','0') as tg_prod_hold_flag,
        |if（regexp_replace(E.tg_prod_coup_unused_flag,',','')='1','1','0') as tg_prod_coup_unused_flag,
        |if（regexp_replace(E.tg_prod_purch_freq_type,',','')='2','1','0') as tg_prod_purch_freq_type_sg,
        |if（regexp_replace(E.tg_prod_purch_freq_type,',','')='1','1','0') as tg_prod_purch_freq_type_fg,
        |if（regexp_replace(E.tg_prod_due_nopurch_flag,',','')='1','1','0') as tg_prod_due_nopurch_flag,
        |if（regexp_replace(E.tg_prod_due_flag_3d,',','')='1','1','0') as tg_prod_due_flag_3d,
        |if（regexp_replace(E.tg_prod_due_flag_1d,',','')='1','1','0') as tg_prod_due_flag_1d
        |from
        |app_data.app_dao_acct_info_d_q A
        |left join app_data.loss_cust_score_feature_info C
        |on  A.capi_acc=C.asset_acct
        |left  join (
        |select
        |id,
        |max(case when lbl_type_no ='SJ01' THEN lbl_val END) as SJ01,
        |max(case when lbl_type_no ='SJ02' THEN lbl_val END) as SJ02,
        |max(case when lbl_type_no ='SJ03' THEN lbl_val END) as SJ03,
        |max(case when lbl_type_no ='SJ04' THEN lbl_val END) as SJ04,
        |max(case when lbl_type_no ='SJ05' THEN lbl_val END) as SJ05,
        |max(case when lbl_type_no ='SJ06' THEN lbl_val END) as SJ06,
        |max(case when lbl_type_no ='SJ07' THEN lbl_val END) as SJ07,
        |max(case when lbl_type_no ='FX01' THEN lbl_val END )as FX01,
        |max(case when lbl_type_no ='FX02' THEN lbl_val END) as FX02,
        |max(case when lbl_type_no ='FX03' THEN lbl_val END )as FX03,
        |max(case when lbl_type_no ='FX04' THEN lbl_val END )as FX04,
        |max(case when lbl_type_no ='SJ08' THEN lbl_val END )as SJ08,
        |max(case when lbl_type_no ='SJ09' THEN lbl_val END )as SJ09,
        |max(case when lbl_type_no ='SJ10' THEN lbl_val END )as SJ10,
        |max(case when lbl_type_no ='SJ11' THEN lbl_val END )as SJ11
        |from sum_data.sum_dao_lbl_custchg_d_q
        |WHERE biz_date = ${inputDt}
        |and id_type= 'a0101'
        |group by id
        |)  B
        |on A.capi_acc = B.id
        |left join lms_data.qw_chg_lbl E
        |on  A.capi_acc=E.asset_acct_general
        |and E.biz_date=${inputDt}
        |left join sum_data.sum_t98_cust_base_info_k F
        |on a.cust_no=F.cust_id
        |and F.biz_date=${inputDt}
        |where A.biz_date=${inputDt} and A.cust_no is not null
      """.stripMargin
    print(allcompany)
    val frame = hiveContext.sql(allcompany)
    val result = frame.rdd.map(x => {
      var cominfo = new StringBuilder()
      val loss_reason = x.getAs[String]("loss_reason")
      val ecif_cd = x.getAs[String]("ecif_cd")
      val capi_acc = x.getAs[String]("capi_acc")
      val cust_no = x.getAs[String]("cust_no")
      val cust_name = x.getAs[String]("cust_name")
      val brch_org_cd = x.getAs[String]("brch_org_cd")
      val brch_org_cd_dev = x.getAs[String]("brch_org_cd_dev")
      val brch_org_cd_sign = x.getAs[String]("brch_org_cd_sign")
      val brch_org_level_cd = x.getAs[String]("brch_org_level_cd")
      val brch_org_level_cd_dev = x.getAs[String]("brch_org_level_cd_dev")
      val brch_org_level_cd_sign = x.getAs[String]("brch_org_level_cd_sign")
      val mgr_cd = x.getAs[String]("mgr_cd")
      val mgr_dev = x.getAs[String]("mgr_dev")
      val mgr_sign = x.getAs[String]("mgr_sign")
      val acc_clas = x.getAs[String]("acc_clas")
      val total_asset =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("total_asset"))
      val total_asset1 = total_asset.bigDecimal

      val net_asset =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("net_asset"))
      val net_asset1 = net_asset.bigDecimal

      val latest_risk_grad = x.getAs[String]("latest_risk_grad")
      val cust_grad = x.getAs[String]("cust_grad")
      val cust_type = x.getAs[String]("cust_type")

      val scr_trdy_20 =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("scr_trdy_20"))
      val scr_trdy_201 = scr_trdy_20.bigDecimal

      val net_asset_avg_20 =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("net_asset_avg_20"))
      val net_asset_avg_201 = net_asset_avg_20.bigDecimal

      val stk_mkt_val =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("stk_mkt_val"))
      val stk_mkt_val1 = stk_mkt_val.bigDecimal

      val total_mkt_val =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("total_mkt_val"))
      val total_mkt_val1 = total_mkt_val.bigDecimal
      val model_score =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("model_score"))
      val model_score1 = model_score.bigDecimal

      val prft_loss_y =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("prft_loss_y"))
      val prft_loss_y1 = prft_loss_y.bigDecimal

      val prft_loss_ly =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("prft_loss_ly"))
      val prft_loss_ly1 = prft_loss_ly.bigDecimal

      val prft_loss_ly2 =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("prft_loss_ly2"))
      val prft_loss_ly21 = prft_loss_ly2.bigDecimal

      val prft_loss_ly3 =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("prft_loss_ly3"))
      val prft_loss_ly31 = prft_loss_ly3.bigDecimal

      val prft_rate_y =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("prft_rate_y"))
      val prft_rate_y1 = prft_rate_y.bigDecimal

      val prft_rate_ly =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("prft_rate_ly"))
      val prft_rate_ly1 = prft_rate_ly.bigDecimal

      val prft_rate_ly2 =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("prft_rate_ly2"))
      val prft_rate_ly21 = prft_rate_ly2.bigDecimal

      val prft_rate_ly3 =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("prft_rate_ly3"))
      val prft_rate_ly31 = prft_rate_ly3.bigDecimal

      val prft_loss_m =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("prft_loss_m"))
      val prft_loss_m1 = prft_loss_m.bigDecimal

      val fm_mkt_val =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("fm_mkt_val"))
      val fm_mkt_val1 = fm_mkt_val.bigDecimal

      val fnd_mkt_val =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("fnd_mkt_val"))
      val fnd_mkt_val1 = fnd_mkt_val.bigDecimal

      val bond_mkt_val =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("bond_mkt_val"))
      val bond_mkt_val1 = bond_mkt_val.bigDecimal

      val cap_bal =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("cap_bal"))
      val cap_bal1 = cap_bal.bigDecimal

      val fin_prod_val =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("fin_prod_val"))
      val fin_prod_val1 = fin_prod_val.bigDecimal

      val cap_in =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("cap_in"))
      val cap_in1 = cap_in.bigDecimal

      val cap_out =  scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("cap_out"))
      val cap_out1 = cap_out.bigDecimal

      val fnd_share_bal2 =  x.getAs[Long]("fnd_share_bal")
      /*val fnd_share_bal = scala.math.BigInt(x.getAs[java.math.BigInteger]("fnd_share_bal"))
      val fnd_share_bal1 = fnd_share_bal.bigInteger*/

      val l1y_purc_same_fnd =  x.getAs[Long]("l1y_purc_same_fnd")

      val l1y_purc_fnd =  x.getAs[Long]("l1y_purc_fnd")

      val l1y_new_stk_allo =  x.getAs[Long]("l1y_new_stk_allo")

      val bjs_flag = x.getAs[String]("bjs_flag")


      val marg_flag = x.getAs[String]("marg_flag")
      val sgt_flag = x.getAs[String]("sgt_flag")
      val hgt_flag = x.getAs[String]("hgt_flag")
      val opt_flag = x.getAs[String]("opt_flag")
      val secu_accet_flag = x.getAs[String]("secu_accet_flag")
      val pro_investor_flag = x.getAs[String]("pro_investor_flag")
      val effect_flag = x.getAs[String]("effect_flag")



      val logn_tims_w =  x.getAs[Long]("logn_tims_w")

      val logn_tims_l4w =  x.getAs[Long]("logn_tims_l4w")

      val logn_tims_l1m =  x.getAs[Long]("logn_tims_l1m")

      val tm = x.getAs[Date]("tm")

      val SJ01 = x.getAs[String]("SJ01")
      val SJ02 = x.getAs[String]("SJ02")
      val SJ03 = x.getAs[String]("SJ03")
      val SJ04 = x.getAs[String]("SJ04")
      val SJ05 = x.getAs[String]("SJ05")
      val SJ06 = x.getAs[String]("SJ06")
      val SJ07 = x.getAs[String]("SJ07")
      val FX01 = x.getAs[String]("FX01")
      val FX02 = x.getAs[String]("FX02")
      val FX03 = x.getAs[String]("FX03")
      val FX04 = x.getAs[String]("FX04")
      val FX05 = x.getAs[String]("FX05")
      val FX06 = x.getAs[String]("FX06")

      val SJ08 = x.getAs[String]("SJ08")
      val SJ09 = x.getAs[String]("SJ09")
      val SJ10 = x.getAs[String]("SJ10")
      val SJ11 = x.getAs[String]("SJ11")
//      val KBS01 = x.getAs[String]("KBS01")
//      val KBS02 = x.getAs[String]("KBS02")
//      val KBS03 = x.getAs[String]("KBS03")
//      val KBS04 = x.getAs[String]("KBS04")
//      val KBS05 = x.getAs[String]("KBS05")
//      val KBS06 = x.getAs[String]("KBS06")
//      val KBS07 = x.getAs[String]("KBS07")
      val  ylj_cust_logo= x.getAs[String]("ylj_cust_logo")
      val  pension_appoint_open_sign= x.getAs[String]("pension_appoint_open_sign")
      val  ind_pension_age_segment_3= x.getAs[String]("ind_pension_age_segment_3")
      val  ind_pension_age_segment_4= x.getAs[String]("ind_pension_age_segment_4")
      val  inv_assets_for_risk_assess_a= x.getAs[String]("inv_assets_for_risk_assess_a")
      val  inv_assets_for_risk_assess_b= x.getAs[String]("inv_assets_for_risk_assess_b")
      val  inv_assets_for_risk_assess_cd= x.getAs[String]("inv_assets_for_risk_assess_cd")

      val  prod_sub_match_result_12= x.getAs[String]("prod_sub_match_result_12")
      val  prod_sub_match_result_3= x.getAs[String]("prod_sub_match_result_3")
      val  tg_prod_cust_flag=x.getAs[String]("tg_prod_cust_flag")
      val  tg_prod_hold_flag=x.getAs[String]("tg_prod_hold_flag")
      val  tg_prod_coup_unused_flag=x.getAs[String]("tg_prod_coup_unused_flag")
      val  tg_prod_purch_freq_type_sg=x.getAs[String]("tg_prod_purch_freq_type_sg")
      val  tg_prod_purch_freq_type_fg=x.getAs[String]("tg_prod_purch_freq_type_fg")
      val  tg_prod_due_nopurch_flag=x.getAs[String]("tg_prod_due_nopurch_flag")
      val  tg_prod_due_flag_3d=x.getAs[String]("tg_prod_due_flag_3d")
      val  tg_prod_due_flag_1d=x.getAs[String]("tg_prod_due_flag_1d")



      val _id = new ID()
      val index = new Index()
      val companyInfo = new ChgeCustIndxDetail()

      _id.set_id(UUID.randomUUID().toString)
      index.setIndex(_id)
      companyInfo.setEcifCd(ecif_cd)
      companyInfo.setLossReason(loss_reason)
      companyInfo.setModelScore(model_score1)
      companyInfo.setCapiAcc(capi_acc)
      companyInfo.setCustNo(cust_no)
      companyInfo.setCustName(cust_name)
      companyInfo.setBrchOrgCd(brch_org_cd)
      companyInfo.setBrchOrgCdDev(brch_org_cd_dev)
      companyInfo.setBrchOrgCdSign(brch_org_cd_sign)
      companyInfo.setBrchOrgLevelCd(brch_org_level_cd)
      companyInfo.setBrchOrgLevelCdDev(brch_org_level_cd_dev)
      companyInfo.setBrchOrgLevelCdSign(brch_org_level_cd_sign)
      companyInfo.setMgrCd(mgr_cd)
      companyInfo.setMgrDev(mgr_dev)
      companyInfo.setMgrSign(mgr_sign)
      companyInfo.setAccClas(acc_clas)
      companyInfo.setTotalAsset(total_asset1)
      companyInfo.setNetAsset(net_asset1)
      companyInfo.setLatestRiskGrad(latest_risk_grad)
      companyInfo.setCustGrad(cust_grad)
      companyInfo.setCustType(cust_type)
      companyInfo.setScrTrdy20(scr_trdy_201)

      companyInfo.setNetAssetAvg20(net_asset_avg_201)
      companyInfo.setStkMktVal(stk_mkt_val1)
      companyInfo.setTotalMktVal(total_mkt_val1)
      companyInfo.setPrftLossY(prft_loss_y1)
      companyInfo.setPrftLossLy(prft_loss_ly1)
      companyInfo.setPrftLossLy2(prft_loss_ly21)
      companyInfo.setPrftLossLy3(prft_loss_ly31)
      companyInfo.setPrftRateY(prft_rate_y1)
      companyInfo.setPrftRateLy(prft_rate_ly1)
      companyInfo.setPrftRateLy2(prft_rate_ly21)
      companyInfo.setPrftRateLy3(prft_rate_ly31)
      companyInfo.setPrftLossM(prft_loss_m1)
      companyInfo.setFmMktVal(fm_mkt_val1)
      companyInfo.setFndMktVal(fnd_mkt_val1)
      companyInfo.setBondMktVal(bond_mkt_val1)
      companyInfo.setCapBal(cap_bal1)
      companyInfo.setFinProdVal(fin_prod_val1)
      companyInfo.setCapIn(cap_in1)
      companyInfo.setCapOut(cap_out1)
      companyInfo.setFndShareBal(fnd_share_bal2)
      companyInfo.setL1yPurcSameFnd(l1y_purc_same_fnd)
      companyInfo.setL1yPurcFnd(l1y_purc_fnd)
      companyInfo.setL1yNewStkAllo(l1y_new_stk_allo)
      companyInfo.setBjsFlag(bjs_flag)
      companyInfo.setMargFlag(marg_flag)
      companyInfo.setHgtFlag(sgt_flag)
      companyInfo.setSgtFlag(hgt_flag)
      companyInfo.setOptFlag(opt_flag)
      companyInfo.setSecuAccetFlag(secu_accet_flag)
      companyInfo.setProInvestorFlag(pro_investor_flag)
      companyInfo.setEffectFlag(effect_flag)
      companyInfo.setLognTimsW(logn_tims_w)
      companyInfo.setLognTimsL4w(logn_tims_l4w)
      companyInfo.setLognTimsL1m(logn_tims_l1m)
      companyInfo.setTm(tm)

      companyInfo.setSJ01(SJ01)
      companyInfo.setSJ02(SJ02)
      companyInfo.setSJ03(SJ03)
      companyInfo.setSJ04(SJ04)
      companyInfo.setSJ05(SJ05)
      companyInfo.setSJ06(SJ06)
      companyInfo.setSJ07(SJ07)
      companyInfo.setFX01(FX01)
      companyInfo.setFX02(FX02)
      companyInfo.setFX03(FX03)
      companyInfo.setFX04(FX04)
      companyInfo.setFX05(FX05)
      companyInfo.setFX06(FX06)

      companyInfo.setSJ08(SJ08)
      companyInfo.setSJ09(SJ09)
      companyInfo.setSJ10(SJ10)
      companyInfo.setSJ11(SJ11)
//      companyInfo.setKBS01(KBS01)
//      companyInfo.setKBS02(KBS02)
//      companyInfo.setKBS03(KBS03)
//      companyInfo.setKBS04(KBS04)
//      companyInfo.setKBS05(KBS05)
//      companyInfo.setKBS06(KBS06)
//      companyInfo.setKBS07(KBS07)
      companyInfo.setYljCustLogo(ylj_cust_logo)
      companyInfo.setPensionAppointOpenSign(pension_appoint_open_sign)
      companyInfo.setIndPensionAgeSegment3(ind_pension_age_segment_3)
      companyInfo.setIndPensionAgeSegment4(ind_pension_age_segment_4)
      companyInfo.setInvAssetsForRiskAssessA(inv_assets_for_risk_assess_a)
      companyInfo.setInvAssetsForRiskAssessB(inv_assets_for_risk_assess_b)
      companyInfo.setInvAssetsForRiskAssessCD(inv_assets_for_risk_assess_cd)
      companyInfo.setProdSubMatchResult12(prod_sub_match_result_12)
      companyInfo.setProdSubMatchResult3(prod_sub_match_result_3)
      companyInfo.setTgProdCustFlag(tg_prod_cust_flag)
      companyInfo.setTgProdHoldFlag(tg_prod_hold_flag)
      companyInfo.setTgProdCoupUnusedFlag(tg_prod_coup_unused_flag)
      companyInfo.setTgProdPurchFreqTypeSg(tg_prod_purch_freq_type_sg)
      companyInfo.setTgProdPurchFreqTypeFg(tg_prod_purch_freq_type_fg)
      companyInfo.setTgProdDueNopurchFlag(tg_prod_due_nopurch_flag)
      companyInfo.setTgProdDueFlag3d(tg_prod_due_flag_3d)
      companyInfo.setTgProdDueFlag1d(tg_prod_due_flag_1d)

      val mapper = new ObjectMapper()

      val indexJson = mapper.writeValueAsString(index) //企业自增编码json串
      val companyInfoJson = mapper.writeValueAsString(companyInfo) //企业信息串

      cominfo = cominfo.append(indexJson + "\n" + companyInfoJson) //json结果串
      cominfo.toString()
    })
    val path = new Path(outputPath) //创建文本输出路径
    val fs = FileSystem.get(URI.create(outputPath), conf)
    if (fs.exists(path)) {
      fs.delete(path, true)
    }
    result.coalesce(splitNum.toInt,true).saveAsTextFile(outputPath)
    sc.stop()
  }
}
